package com.neusoft.neuiotms.support.listener;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import com.neusoft.bizcore.web.bean.LogBean;
import com.neusoft.bizcore.web.utils.JsonUtils;
import com.neusoft.neuiotms.support.service.LogService;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
@ConditionalOnClass(KafkaTemplate.class)
public class KafkaLogListener {

    @Autowired
    private LogService logService;

    @KafkaListener(groupId = "${spring.kafka.consumer.group-id}", topics = "${support.log.kafka-topic}")
    public void receiveTopic(final ConsumerRecord<String, String> record) {
        if (KafkaLogListener.log.isDebugEnabled()) {
            KafkaLogListener.log.debug("{}", record);
        }
        try {
            final LogBean bean = JsonUtils.jsonToPojo(record.value(), LogBean.class);

            this.logService.save(bean);
        } catch (final Throwable e) {
            KafkaLogListener.log.error("", e);
        }
    }
}
